#include "iostreambase.h"
#include "ioeventloop.h"

uvNameSpaceBegin
//////////////////////////////////////////////////////////////////////////////////
static void on_readcb(struct bufferevent *bev, void *arg)
{
	io_streambase_t* p = (io_streambase_t*)arg;
	ptrnull_return(p)
	p->io.stream.ulReadLen = bufferevent_read(bev, p->io.stream.pReadPtr, p->io.stream.ulReadLen);
	if (!(p->readlive & STREAM_READING))
		bufferevent_disable(p->bev, EV_READ);
	p->stream_fun_readcb(p, p->io.stream.ulReadLen);
}
static void on_writecb(struct bufferevent *bev, void *arg)
{
	io_streambase_t* p = (io_streambase_t*)arg;
	ptrnull_return(p)
	size_t wlen = evbuffer_get_length(bufferevent_get_output(bev));
	if (!(p->writealive & STREAM_WRITEING))
		bufferevent_disable(p->bev, EV_WRITE);
	p->stream_fun_writecb(p, wlen);
}
static void on_eventcb(struct bufferevent *bev, short sEvent, void *arg)
{
	io_streambase_t* p = (io_streambase_t*)arg;
	ptrnull_return(p)
	
	if (sEvent & BEV_EVENT_READING) {
		p->state = STREAM_READ;
	}

	if (sEvent & BEV_EVENT_WRITING) {
		p->state = STREAM_WRITE;
	}

	if (sEvent & BEV_EVENT_ERROR) {
		p->stream_fun_eventcb(p, STREAM_EVENT_ERROR);
	}

	if (sEvent & BEV_EVENT_EOF) {
		p->stream_fun_eventcb(p, STREAM_EVENT_EOF);
	}

	if (sEvent &  BEV_EVENT_TIMEOUT) {
		p->stream_fun_eventcb(p, STREAM_EVENT_TIMEOUT);
	}

}
static void MultiThread(io_streambase_t* pStream) {

	if (pStream->thread) {
		io_thread_t* p = geteventloop();
		pStream->bev->ev_base = p->eventloop->loop;
		event_base_set(p->eventloop->loop, &pStream->bev->ev_read);
		event_base_set(p->eventloop->loop, &pStream->bev->ev_write);
	}
}

int	io_streambase_init(io_streambase_t* pStream, io_tcp_i* pTcp, int iReadTimeOut, int iWriteTimeOut) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	evutil_make_socket_nonblocking(pTcp->sock);
	pStream->bev = bufferevent_socket_new(pTcp->eventloop->loop, pTcp->sock, BEV_OPT_DEFER_CALLBACKS | BEV_OPT_THREADSAFE);

	pStream->input = bufferevent_get_input(pStream->bev);
	pStream->output = bufferevent_get_output(pStream->bev);

	pStream->bev->timeout_read.tv_sec = iReadTimeOut;
	pStream->bev->timeout_read.tv_usec = 0;
	pStream->bev->timeout_write.tv_sec = iWriteTimeOut;
	pStream->bev->timeout_write.tv_usec = 0;

	pStream->readlive	= 0;
	pStream->writealive = 0;
	pStream->thread		= false;

	bufferevent_setcb(pStream->bev, on_readcb, on_writecb, on_eventcb, pStream);
	bufferevent_disable(pStream->bev, EV_READ|EV_WRITE | EV_PERSIST | EV_ET);
	bufferevent_set_timeouts(pStream->bev, &pStream->bev->timeout_read, &pStream->bev->timeout_write);
	return 1;
}


int	io_streambase_read_timeout(io_streambase_t* pStream, int iReadTimeOut) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	pStream->bev->timeout_read.tv_sec = iReadTimeOut;
	pStream->bev->timeout_read.tv_usec = 0;
	return 1;
}
int	io_streambase_wrtie_timeout(io_streambase_t* pStream, int iWriteTimeOut) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	pStream->bev->timeout_write.tv_sec = iWriteTimeOut;
	pStream->bev->timeout_write.tv_usec = 0;
	return 1;
}

int	io_streambase_write(io_streambase_t* pStream, unsigned long ulMinSize){

	ptrnull_value(pStream, IO_VALUE_ERROR);
	MultiThread(pStream);

	pStream->io.stream.ulWriteLen = ulMinSize;
	bufferevent_setwatermark(pStream->bev, EV_WRITE, ulMinSize, pStream->io.stream.ulWriteLen);
	bufferevent_write(pStream->bev, pStream->io.stream.pWritePtr, ulMinSize);
	return bufferevent_enable(pStream->bev, EV_WRITE);
}

int	io_streambase_read(io_streambase_t* pStream, unsigned long ulMinSize) {

	//pStream->writealive = STREAM_WRITEING;
	//pStream->readlive = STREAM_READING;
	ptrnull_value(pStream, IO_VALUE_ERROR);
	MultiThread(pStream);
	bufferevent_setwatermark(pStream->bev, EV_READ, ulMinSize, pStream->io.stream.ulReadLen);
	return bufferevent_enable(pStream->bev, EV_READ);
}

#define MAX_READ_SIZE 0
int	io_streambase_read_some(io_streambase_t* pStream) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	MultiThread(pStream);
	bufferevent_setwatermark(pStream->bev, EV_READ, MAX_READ_SIZE, pStream->io.stream.ulReadLen);
	return bufferevent_enable(pStream->bev, EV_READ);
}
int	io_streambase_write_some(io_streambase_t* pStream) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	MultiThread(pStream);
	bufferevent_write(pStream->bev, pStream->io.stream.pWritePtr, pStream->io.stream.ulWriteLen);
	bufferevent_setwatermark(pStream->bev, EV_WRITE, pStream->io.stream.ulWriteLen, pStream->io.stream.ulWriteLen);
	return bufferevent_enable(pStream->bev, EV_WRITE);
}
int	io_streambase_close(io_streambase_t* pStream) {

	ptrnull_value(pStream, IO_VALUE_ERROR);
	bufferevent_free(pStream->bev);
	return 1;
}
uvNameSpaceEnd